package net.jacqeush.mini.rpc.client.manage;

import com.aliyun.opensearch.sdk.dependencies.com.google.common.util.concurrent.ThreadFactoryBuilder;

import lombok.extern.slf4j.Slf4j;
import net.jacqeush.mini.rpc.codec.model.Request;
import net.jacqeush.mini.rpc.codec.model.Response;


import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

import java.util.concurrent.locks.ReentrantLock;

/**
 * @author Jacquesh
 * @version 1.0
 * @date 2020/10/17 3:37
 */
@Slf4j
public class RpcFuture implements Future<Object> {

    private Request request;
    private Response response;
    private long startTime;
    private static final long TIME_THRESHOLD = 5000;
    private List<RpcCallback> callbacks = new ArrayList<RpcCallback>();
    private FutureSync sync;
    private ReentrantLock lock = new ReentrantLock();
    ThreadFactory namedFactory = new ThreadFactoryBuilder().setNameFormat("future-pool-%d").build();
    ExecutorService executor = new ThreadPoolExecutor(16, 16,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(Integer.MAX_VALUE),
            namedFactory, new ThreadPoolExecutor.AbortPolicy());

    public RpcFuture(Request request) {
        this.request = request;
        this.startTime = System.currentTimeMillis();
        this.sync = new FutureSync();

    }

    public void done(Response response) {
        this.response = response;
        boolean success = sync.release(1);
        if(success) {
            invokeCallbacks();
        }
        // 整体rpc调用的耗时
        long costTime = System.currentTimeMillis() - startTime;
        if(TIME_THRESHOLD < costTime) {
            log.warn("the rpc response time is too slow, request id = " + this.request.getRequestId() + " cost time: " + costTime);
        }
    }

    /**
     * 	依次执行回调函数处理
     */
    private void invokeCallbacks() {
        lock.lock();
        try {
            for(final RpcCallback callback : callbacks) {
                runCallback(callback);
            }
        } finally {
            lock.unlock();
        }
    }

    private void runCallback(RpcCallback callback) {
        final Response response = this.response;
        executor.submit(()-> {

                if(response.getThrowable() == null) {
                    callback.success(response.getResult());
                } else {
                    callback.failure(response.getThrowable());
                }

        });
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isDone() {
        return sync.isDone();
    }

    @Override
    public Object get() {
        this.sync.acquire(-1);
       if(this.response!=null){
           return this.response.getResult();
       }else {
           return null;
       }
    }

    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException {
        boolean success = sync.tryAcquireNanos(-1, unit.toNanos(timeout));
        if(success) {
            if(this.response != null) {
                return this.response.getResult();
            } else {
                return null;
            }
        } else {
            throw new RuntimeException("timeout excetion requestId: "
                    + this.request.getRequestId()
                    + ",className: " + this.request.getClassName()
                    + ",methodName: " + this.request.getMethodName());
        }
    }

}
